package com.bungarus;

import com.bungarus.busi.AccessBusiProcessing;
import com.bungarus.busi.AccessRemedy;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.config.ChannelAccessInitializer;
import com.bungarus.model.*;
import com.bungarus.utils.IPAddressUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelHandler;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.protocol.HTTP;
import org.apache.http.protocol.HttpContext;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.net.UnknownHostException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;

/**
 * Entry point of Access Tier
 * 1. Retrieve transfer / busi cluster nodes according to specified cluster calculated from access-cluster-id % transfer-ids
 * 2. Save the transfer / busi nodes into an array
 * 3. calculate the transfer / busi node the access can connect to
 *
 * Created by tang on 2018/11/11.
 */
public class Access extends AbstractBungarusServerWithMultipleClients {
    private Logger logger = Logger.getLogger(Access.class);

    private static final String RM_CONNECT_TIMEOUT = "bungarus.rm.timeout";
    private static final String MSGIDSCOPE_ENDPOINT = "bungarus.rm.msgidscope.endpoint";
    private static final String MSGID_SEPERATOR = ",";
    private static final String SUPPORT_SSL = "ssl";

    public static final String TAG = "access";
    /**
     * clusters connected to as client
     */
    private static final String[] CLUSTER_TAGS = {CLUSTER_TAG_TRANSFER, CLUSTER_TAG_BUSI};

    private static final String RUNTIME = "local";

    public static final int PORT_80 = 9080;
    public static final int PORT_8080 = 9980;
    public static final int PORT_7700 = 7700;
    public static final int PORT_7800 = 7800;
    /**
     * session container for maintaining the client connection session
     */
    private static final SessionContainer sessionContainer = new SessionContainer();

    private SnowflakeIdMaker idMaker = null;
    private final boolean fromLocal;
    /**
     * support ssl or not, default supports
     */
    private boolean ssl;

    public Access(boolean fromLocal) {
        this(fromLocal, true);
    }

    public Access(boolean fromLocal, boolean supportSSL) {
        super();
        this.ssl = supportSSL;
        this.fromLocal = fromLocal;
    }

    public static SessionContainer getSessionContainer(){
        return sessionContainer;
    }

    public static void main(String[] args) throws Exception{
        int[] messagePorts = {PORT_80, PORT_8080};
        int[] heartbeatPorts = {PORT_7700, PORT_7800};

        boolean local = Boolean.parseBoolean(System.getProperty(RUNTIME, "true"));

        Access access = new Access(local, Boolean.parseBoolean(System.getProperty(SUPPORT_SSL, "true")));
        //as a server
        access.bind(System.getProperty(AbstractBungarusServer.HOST), messagePorts, heartbeatPorts);
        access.initClusterAndConnectRemoteNode();
        access.initMsgIDScope();
        access.initRemedy();
        access.listen(null);
    }

    private void initMsgIDScope() throws InterruptedException {
        //1. init httpClient
        CloseableHttpClient client = initClient();

        //2. invoke the ResourceManager service to get the message-id scope
        String rmEndPoint = AccessBusiProcessing.springContext.getEnvironment().getProperty(MSGIDSCOPE_ENDPOINT);
        if(null == rmEndPoint || rmEndPoint.length() <= 0){
            throw new IllegalStateException("Endpoint of service for getting message id scope needed.");
        }
        StringBuilder sb = new StringBuilder(rmEndPoint);
        if(!rmEndPoint.endsWith("/")){
            sb.append("/");
        }
        sb.append(this.getNodeInfo().get().getClusterIDBelongsTo());
        sb.append(MSGID_SEPERATOR);
        sb.append(this.getNodeInfo().get().getNodeId());

        HttpGet getMethod = new HttpGet(sb.toString());
        int times = 0;
        boolean done = false;
        do {
            try {
                CloseableHttpResponse response = client.execute(getMethod);
                HttpEntity entity = response.getEntity();
                if (entity.getContentType().getValue().contains("application/json")) {
                    String strEntity = EntityUtils.toString(entity, "utf-8");

                    ObjectMapper objectMapper = new ObjectMapper();
                    MessageIDScope scope = objectMapper.readValue(strEntity, MessageIDScope.class);
                    idMaker = new SnowflakeIdMaker(scope.getWorker(), scope.getDc());
                    done = true;
                }
            }
            catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(2000);
            }
            finally {
                times++;
            }
        }while(!done && times < 3);
    }

    private CloseableHttpClient initClient() {
        int timeout = Integer.parseInt(AccessBusiProcessing.springContext.getEnvironment().getProperty(RM_CONNECT_TIMEOUT));

        PoolingHttpClientConnectionManager clientConnectionManager = new PoolingHttpClientConnectionManager();
        clientConnectionManager.setMaxTotal(Integer.MAX_VALUE);
        clientConnectionManager.setDefaultMaxPerRoute(Integer.MAX_VALUE);

        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectionRequestTimeout(timeout)
                .setConnectTimeout(timeout * 3)
                .setSocketTimeout(timeout * 3)
                .build();

        CloseableHttpClient client = HttpClients.custom()
                .setKeepAliveStrategy((HttpResponse response, HttpContext context) -> {
                    HeaderElementIterator it = new BasicHeaderElementIterator(response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                    while (it.hasNext()) {
                        HeaderElement he = it.nextElement();
                        String param = he.getName();
                        String value = he.getValue();
                        if (value != null && param.equalsIgnoreCase
                                ("timeout")) {
                            return Long.parseLong(value) * 1000;
                        }
                    }
                    return 5 * 1000;
                })
                .setConnectionManager(clientConnectionManager)
                .setDefaultRequestConfig(requestConfig)
                .build();

        return client;
    }

    @Override
    public void initRemedy() throws Exception {
        AccessRemedy.start(this);
    }

    @Override
    public CopyOnWriteArrayList<NodeInfo> calcRemoteServerNodes(CopyOnWriteArrayList<NodeInfo> originList) {
        if(null != originList && originList.size() > 0 &&
                originList.get(0).getClusterBelongsTo().equalsIgnoreCase(CLUSTER_TAG_BUSI)) {
                //only connect to normal nodes live in busi cluster
            List<NodeInfo> queue = originList.stream().filter(nodeInfo -> !nodeInfo.isHighPerformance()).collect(Collectors.toList());
            CopyOnWriteArrayList<NodeInfo> result = new CopyOnWriteArrayList<>(queue);
            queue.clear();
            return result;
        }
        return originList;
    }

    @Override
    public ChannelHandler getChildHandlerInitializer() {
        return new ChannelAccessInitializer(this);
    }

    @Override
    public MongoTemplate getMongoTemplate4FailedRetry() {
        return (MongoTemplate) AccessBusiProcessing.springContext.getBean("mongoTemplate");
    }

    @Override
    public String getTag() {
        return Access.TAG;
    }

    @Override
    public String tagPortString4Connection() {
        return "(smart|7700,busi|7800)";
    }

    public String[] getRemoteNodeTags() { return CLUSTER_TAGS; }

    public SnowflakeIdMaker getIdMaker() {
        return this.idMaker;
    }

    public boolean sslEnabled() {
        return this.ssl;
    }

    /**
     * Set WAN ip for access node in extra information
     * @return
     */
    @Override
    public String getExtraInfo() {
        try {
            return fromLocal ? IPAddressUtils.getLocalHostLANAddress().getHostAddress() : IPAddressUtils.getLocalHostWANAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return "";
    }
}
